/* file: real_node.cc */

#include <string.h>
#include <errno.h>
#include "real_node.h"

/**********************************
class:	RealNode
method:	initRealNode
**********************************/
void
RealNode::initRealNode()
{
    Set set;			// nodes with same IP and different level
    Set::size_type ix;

    cgtree.loadCGImage("cgtree.conf");
    //cgtree.printCGMap();
    ip = cgtree.getIP();
    status = STAT_NORMAL;
    //task.clear();
    cgtree.indexIPNode(ip, &set);
    for (ix = 0; ix != set.size(); ++ix)
    {
	cgtree.codeCGImage(set[ix].ip, set[ix].level, str);
	nodes[set[ix].level].initCGNode(set[ix].ip, set[ix].level, cgtree.getRoot(), cgtree.getDepth(), str);
	/* node at STAT_INIT */
    }
    
#ifdef DEBUG
    printf("RealNode init complete\n");
#endif
}

void *thread_recv(void *arg)
{
    return ((RealNode *)arg)->recvMessage(NULL);
}
void *thread_recvfile(void *arg)
{
    return ((RealNode *)arg)->recvFile(NULL);
}
void *thread_recvhb(void *arg)
{
    return ((RealNode *)arg)->recvHeartBeat(NULL);
}
void *thread_hb(void *arg)
{
    return ((RealNode *)arg)->heartBeat(NULL);
}
void *thread_recvhandle(void *arg)
{
    return ((RealNode *)arg)->recvMesThread(NULL);
}
void *thread_filehandle(void *arg)
{
    ((RealNode *)(((fileArg *)arg)->REAL_ptr))->recvFileThread(((fileArg *)arg)->connfd, ((fileArg *)arg)->ip);
    free((fileArg *)arg);
    pthread_exit((void *)1);
}
void *thread_findServer(void *arg)
{
    return ((RealNode *)arg)->findServer();
}

/**********************************
class:	RealNode
method:	run
**********************************/
void
RealNode::run()
{
    pthread_t tid_recv, tid_recvfile, tid_recvhb, tid_sendhb, tid_findrs;
    void *tret;
    Set set;			// nodes with same IP and different level
    Set::size_type ix;

#ifdef RUN
    printf("Host %s start\n", ip.c_str());
#endif

    if (pthread_create(&tid_recv, NULL, thread_recv, (void *)this) != 0)
    {
	perror("pthread_create() error ");
	exit(1);
    }
    if (pthread_create(&tid_recvfile, NULL, thread_recvfile, (void *)this) != 0)
    {
	perror("pthread_create() error ");
	exit(1);
    }
    
    sleep(1);
    cgtree.indexIPNode(ip, &set);
    for (ix = 0; ix != set.size(); ++ix)
    {
	nodes[set[ix].level].registerCGNode();
    }
    
    if (pthread_create(&tid_recvhb, NULL, thread_recvhb, (void *)this) != 0)
    {
	perror("pthread_create() error ");
	exit(1);
    }
    if (pthread_create(&tid_sendhb, NULL, thread_hb, (void *)this) != 0)
    {
	perror("pthread_create() error ");
	exit(1);
    }

    if (pthread_create(&tid_findrs, NULL, thread_findServer, (void *)this) != 0)
    {
	perror("pthread_create() error ");
	exit(1);
    }


    if (pthread_join(tid_recv, &tret) != 0)
    {
	perror("pthread_join() error ");
	exit(1);
    }
    if (pthread_join(tid_recvhb, &tret) != 0)
    {
	perror("pthread_join() error ");
	exit(1);
    }
    if (pthread_join(tid_sendhb, &tret) != 0)
    {
	perror("pthread_join() error ");
	exit(1);
    }
    if (pthread_join(tid_findrs, &tret) != 0)
    {
	perror("pthread_join() error ");
	exit(1);
    }
}

/**********************************
class:	RealNode
method:	recvMesThread
**********************************/
void*
RealNode::recvMesThread(void *arg)
{
    MessageType type;
    int r_level, s_level;
    string s_ip;
    char content[BUFF_MAXSIZE];
    map<int, CGNode>::iterator map_it;
    
    r_level = recv_st.getRecvLevel();
    type = recv_st.getType();
    s_ip = recv_st.getSender();
    s_level = recv_st.getSendLevel();
    recv_st.getContent(content);

#ifdef DEBUG
    printf("Level %d get %d from %s(%d)\n", r_level, type, s_ip.c_str(), s_level);
#endif
    
    if (r_level == -1)
    {
	CGNodeStatus stat;
	if (type == NODE_PAUSE)
	{
	    status = STAT_PAUSE;
	    map_it = nodes.begin();
	    while (map_it != nodes.end())
	    {
		if ((stat = map_it->second.getStatus()) != OFFLINE)
		{
		    while (stat != STAT_NORMAL);
		    if (stat == STAT_NORMAL)
		    {
			map_it->second.setStatus(STAT_PAUSE);
#ifdef RUN
			printf("Level %d -> STAT_PAUSE\n", map_it->first);
#endif
		    }
		}
		++map_it;
	    }
	}
	else if (type == NODE_WORK)
	{
	    //if (this->ip != this->getRoot())
	    {
		status = STAT_NORMAL;
		map_it = nodes.begin();
	    	while (map_it != nodes.end())
	    	{
			if (map_it->second.getStatus() == STAT_PAUSE)
			{
			    cgtree.codeCGImage(this->ip, map_it->first, content);
			    map_it->second.initCGNode(this->ip, map_it->first, cgtree.getRoot(), cgtree.getDepth(), content);
			    map_it->second.registerCGNode();
			}
			++map_it;
	    	}
	    }
	    /*else
	    {
		Correspond sendI(TCP, RECV_PORT);
		nodes[0].codeCGImage(content);
		puts(content);
		sendI.setHead(this->ip, -1, s_ip, -1);
		sendI.sendMessage(RECOVER_IMAGE, content);
	    }*/
	}
	return ((void *)0);
    }

    if (status == STAT_PAUSE) return ((void *)1);
    
    if (type == RECOVER_RECONSTRUCT)
    {
	printf("%d", status);
	if (status == STAT_NORMAL)
	{
	    if (!nodes.count(r_level))
		nodes[r_level].initCGNode(this->ip, r_level, cgtree.getRoot(), cgtree.getDepth(), "");
	    else if (nodes[r_level].getStatus() == OFFLINE)
		nodes[r_level].setStatus(STAT_NORMAL);
	}
    }
    
    if (type == RECOVER_RECONFIG)
    {
	if (status == STAT_NORMAL)
	{
	    map_it = nodes.begin();
	    while (map_it != nodes.end())
	    {
		if (map_it->first < r_level && map_it->second.getStatus() > OFFLINE)
		{
		    map_it->second.setStatus(OFFLINE);
#ifdef RUN
		    printf("Level %d be OFFLINE\n", map_it->first);
#endif
		}
		++map_it;
	    }
	}
    }

#ifdef DEBUG
    printf("Level %d at %d\n", r_level, nodes[r_level].getStatus());
#endif
    
    if (nodes[r_level].getStatus() != OFFLINE)
    {
	//printf("%d\n", recv_st.getType());
	nodes[r_level].recvMessage(type, s_ip, s_level, content);
    }
    pthread_exit((void *)1);
}

/**********************************
class:	RealNode
method:	recvMessage
param:	connfd - file descriptor
	ip - sender
**********************************/
void*
RealNode::recvFileThread(int connfd, char *ip)
{
    FILE *fp;
    char filename[30], buf[BUFF_MAXSIZE];
    int n;
    int lv;

    if ((fp = fopen(ip, "wb+")) < 0)
    {
	perror("fopen() error");
	pthread_exit((void *)-1);
    }
    while ((n = recv(connfd, buf, 4096, 0)) > 0)
    {
	if ((n = fwrite(buf, 1, n, fp)) > 0);
    }
    close(connfd);

    rewind(fp);
    fscanf(fp, "%d", &lv);		// sender's level		
    fclose(fp);

#ifdef DEBUG
    printf("recv rs_list from %s(%d)\n", ip, lv);
#endif
	
    sprintf(filename, "%s_%d", ip, lv);
    rename(ip, filename);
    
    --lv;		// receiver's level
    nodes[lv].recvFile(filename, ip);
}

/**********************************
class:	RealNode
method:	recvMessage
**********************************/
void*
RealNode::recvMessage(void *arg)
{
    int listenfd;
    pthread_t tid_recvhandle;
    char str1[BUFF_MAXSIZE];
   
    /* state transformation correspond */
    recv_st.setSock(TCP);
    recv_st.setPort(RECV_PORT);
    listenfd = recv_st.listenPort();

#ifdef DEBUG
    printf("recvMessage\n");
#endif
    while (1)
    {
	if (recv_st.recvMessage(listenfd) == 0)
	{
	    if (pthread_create(&tid_recvhandle, NULL, thread_recvhandle, (void *)this) != 0)
	    {
		perror("pthread_create() error ");
		exit(1);
	    }
	}
    }
    pthread_exit((void *)0);
}

/**********************************
class:	RealNode
method:	recvFile
**********************************/
void*
RealNode::recvFile(void *arg)
{
    int listenfd, connfd;
    struct sockaddr_in servaddr, cliaddr;
    socklen_t len;
    pthread_t tid_filehandle;
    //fileArg fa;

#ifdef DEBUG
    printf("recvFile\n");
#endif
    
    if((listenfd = socket(AF_INET,SOCK_STREAM,0)) < 0){
        printf("socket() error\n");
        pthread_exit((void *)-1);
    }
    bzero(&servaddr,sizeof(servaddr));
    servaddr.sin_family = AF_INET;
    servaddr.sin_addr.s_addr = htonl(INADDR_ANY);
    servaddr.sin_port = htons(FILE_PORT);
    if((bind(listenfd,(sockaddr*)&servaddr,sizeof(servaddr))) < 0){
        printf("%s bind() error\n",ip.c_str());
        pthread_exit((void *)-1);
    }
    listen(listenfd, 25);
    
    while (1)
    {
	len = sizeof(cliaddr);
	connfd = accept(listenfd, (struct sockaddr *)&cliaddr, &len);
	fileArg *fa = (fileArg *)malloc(sizeof(fileArg));
	fa->connfd = connfd;
	inet_ntop(AF_INET, &cliaddr.sin_addr, fa->ip, sizeof(fa->ip));
	fa->REAL_ptr = (void *)this;
	if (pthread_create(&tid_filehandle, NULL, thread_filehandle, (void *)fa) < 0)
	{}
    }
}

/**********************************
class:	RealNode
method:	recvHeartBeat
**********************************/
void*
RealNode::recvHeartBeat(void *arg)
{
    int listenfd;
    char str[BUFF_MAXSIZE];
    
    /* heartbeat correspond */
    recv_hb.setSock(UDP);
    recv_hb.setPort(HB_PORT);
    listenfd = recv_hb.listenPort();
    
    while (1)
    {
	recv_hb.recvMessage(listenfd);
	if (recv_hb.getType() == HEART_BEAT)
	{
	    recv_hb.getContent(str);
	    nodes[recv_hb.getRecvLevel()].recvHeartBeat(recv_hb.getType(), recv_hb.getSender(), str);
	}
    }
    pthread_exit((void *)0);
}

/**********************************
class:	RealNode
method:	sendHeartBeat
**********************************/
void*
RealNode::heartBeat(void *arg)
{
    map<int, CGNode>::iterator node_it = nodes.begin();

#ifdef DEBUG
    printf("heartBeat start\n");
#endif
    while (1)
    {
	sleep(1);	// for each CGNode, if hb_sec is time up, send heart_beat
	for (node_it = nodes.begin(); node_it != nodes.end(); ++node_it)
	    node_it->second.checkTimeout();
	//printf("checkTimeOut\n");
	for (node_it = nodes.begin(); node_it != nodes.end(); ++node_it)
	    node_it->second.sendHeartBeat(load);
	//printf("sendHB\n");
	if (task.size() > 0)
	{
	    for (map<string, int>::iterator task_it = task.begin(); task_it != task.end(); ++task_it)
	    {
		if (++(task_it->second) >= 10)
		    rootServeCli((char *)(task_it->first).c_str());
	    }
	}
    }
    pthread_exit((void *)0);
}
    
/*void *test(void *arg)
{
    Set set;
    cgtree.loadCGImage("./cgtree.conf");
    ip = cgtree.getIP();
    //cgtree.printCGMap();
    cgtree.codeCGImage(ip, 0, str);
    puts(str);
    CGImage image;
    image.buildCGImage(str);
    image.printCGMap();
    
    return 0;
}*/

void RealNode::rootServeCli(char *str)
{
    Correspond send_service(TCP,SERVICE_PORT);
    char req_ip[20];
    char filename[30];
    char server[20];

    sscanf(str,"%s %s",req_ip,filename);
    task[str] = 0;
    
    if(nodes[0].has_source(filename)) {
	if(nodes[0].match_ip(req_ip)) {
	    if(getDepth() - 2 == 0) {
		if(nodes[0].findFit(filename,server)){
		    sprintf(str,"%s %d %s ",server,1,filename);

                    send_service.setHead(ip,0,req_ip,-1);
		    send_service.sendMessage(ANSWER_REQ,str);
                }
		else{
#ifdef DEBUG
		    printf("FIND_FIT ERROR\n");
#endif
		    //pthread_exit((void*)-1);
		}
	    }
	    else{
		if(nodes[0].sendFirstChild(str) < 0){
		    printf("node[0] send first-child error\n");
								//pthread_exit((void*)-1);
		}
							printf("send to first child\n");
						}
					}
		else{
						if(nodes[0].findFit(filename,server)){
                                                        sprintf(str,"%s %d %s ",server,getDepth()-1,filename);
                                                        send_service.setHead(ip,0,req_ip,-1);
                                                        send_service.sendMessage(ANSWER_REQ,str);
                                                }
                                                else{
                                                        printf("FIND_FIT ERROR\n");
                                                        //pthread_exit((void*)-1);
                                                }
					}
				}
				else{		
					send_service.setHead(ip,0,req_ip,-1);		
					send_service.sendMessage(CANT_FIND,str);
					task.erase(str);
				}
    
}

void* RealNode::findServer(){
	Correspond send_service(TCP,SERVICE_PORT);
	Correspond send_admin(TCP,ADMIN_PORT);
	Correspond send(TCP,RSMESS_PORT);
	char str[BUFF_MAXSIZE];
	char req_ip[20];
	char filename[30];
	char server[20];
	char father[20];
	int listenfd;
	int recvLevel;	

	recv_rs.setSock(TCP);
	recv_rs.setPort(RSMESS_PORT);
	listenfd = recv_rs.listenPort();

	while(1){
		recv_rs.recvMessage(listenfd);
		recvLevel = recv_rs.getRecvLevel();
#ifdef DEBUG
		printf("Receive message from %s[%d]\n",recv_rs.getSender().c_str(),recv_rs.getSendLevel());
#endif
		recv_rs.getContent(str);
		// printf("%s\n",str);
		sscanf(str,"%s %s",req_ip,filename);
		if(recvLevel == 0){
		        while (nodes[0].getStatus() != STAT_NORMAL);
			if(recv_rs.getType() == CLI_REQ){
			    //printf("receive CLI_REQ\n");
			    task[str] = 0;
				if(nodes[0].has_source(filename)){
					//if(nodes[0].match_ip(req_ip)){
						if(getDepth() - 2 == 0){
							if(nodes[0].findFit(filename,server)){
                                        			sprintf(str,"%s %d %s ",server,1,filename);
								
								//printf("str = %s\n",str);

                                        			send_service.setHead(ip,0,req_ip,-1);
                                        			send_service.sendMessage(ANSWER_REQ,str);
                                			}
							else{
								printf("FIND_FIT ERROR\n");
								pthread_exit((void*)-1);
							}
						}
						else{
							if(nodes[0].sendFirstChild(str) < 0){
								printf("node[0] send first-child error\n");
								pthread_exit((void*)-1);
							}
						}
					//}
					/*else{
						if(nodes[0].findFit(filename,server)){
                                                        sprintf(str,"%s %d %s ",server,getDepth()-1,filename);
                                                        send_service.setHead(ip,0,req_ip,-1);
                                                        send_service.sendMessage(ANSWER_REQ,str);
                                                }
                                                else{
                                                        printf("FIND_FIT ERROR\n");
                                                        pthread_exit((void*)-1);
                                                }
					}*/
				}
				else{		
					send_service.setHead(ip,0,req_ip,-1);			
					send_service.sendMessage(CANT_FIND,str);
					task.erase(str);
				}
			}
			else if(recv_rs.getType() == FIND_SUCCESS){
				#ifdef RUN
				printf("recv FIND_SUCCESS from %s[%d]\n",recv_rs.getSender().c_str(),recvLevel+1);
				#endif
				nodes[0].erase(str);
			}
			else if(recv_rs.getType() == FIND_FAIL){
				#ifdef RUN
                                printf("recv FIND_FAIL from %s[%d]\n",recv_rs.getSender().c_str(),recvLevel+1);
                                #endif
				if(nodes[0].sendNextChild(str) < 0){
					nodes[0].erase(str);
//					sleep(20);
					send_service.setHead(ip,0,req_ip,-1);
					send_service.sendMessage(WAIT_REQ,str);
				}
			}
			else if(recv_rs.getType() == SOURCE_REQ){
				sprintf(filename,"%s%c","All_Source",'\0');
				nodes[0].mapToFile(filename);
				nodes[0].sendFile(recv_rs.getSender().c_str(),filename);
				printf("send file to %s[%d] complete \n",recv_rs.getSender().c_str(),-2);
			}
			else if(recv_rs.getType() == LOAD_REQ){
				nodes[0].mapToStr(str);
				send_admin.setHead(ip,0,recv_rs.getSender(),-2);
				send_admin.sendMessage(ANSWER_LOAD,str);
				printf("send str[] to %s[%d] complete \n",recv_rs.getSender().c_str(),-2);
			}
			else if(recv_rs.getType() == ADD_RS){
                                nodes[recvLevel].addIndex(req_ip,filename);
                             //   nodes[recvLevel].getFather(father);
                                send_admin.setHead(ip,recvLevel,"59.77.10.117",-2);
                                send_admin.sendMessage(ADD_COMPLETE,"");
                        }
                        else if(recv_rs.getType() ==DELETE_RS){
                                nodes[recvLevel].deleteIndex(req_ip,filename);
                             //   nodes[recvLevel].getFather(father);
                                send_admin.setHead(ip,recvLevel,"59.77.10.117",-2);
                                send_admin.sendMessage(DELETE_COMPLETE,"");
                        }
			else if(recv_rs.getType() == DO_TASK){
			    task.erase(str);
			}
			else{
				printf("node[0] Receive a unknown message\n");
				pthread_exit((void*)-1);
			}
		}
		else if(recvLevel == getDepth() - 2){
		        while (nodes[recvLevel].getStatus() != STAT_NORMAL);
			if(recv_rs.getType() == FORWARD_REQ){
				if(nodes[recvLevel].has_source(filename)){
					//if(nodes[recvLevel].match_ip(req_ip)){
						nodes[recvLevel].sendSuccess(str);
						if(nodes[recvLevel].findFit(filename,server)){
							/* inform root of doing this task */
							Correspond send_task(TCP, RSMESS_PORT);
							send_task.setHead(ip, recvLevel, getRoot(), 0);
							send_task.sendMessage(DO_TASK, str);
                                                	
							sprintf(str,"%s %d %s ",server,recvLevel+1,filename);
                                                        send_service.setHead(ip,recvLevel,req_ip,-1);
                                                        send_service.sendMessage(ANSWER_REQ,str);
						}
						else{
							printf("nodes[%d] FIND_FIT error\n",recvLevel);
							pthread_exit((void*)-1);
						}
					//}
					/*else
						nodes[recvLevel].sendFail(str);*/
				}
				else
					nodes[recvLevel].sendFail(str);
			}
			else if(recv_rs.getType() == ADD_RS){
                                nodes[recvLevel].addIndex(req_ip,filename);
                                nodes[recvLevel].getFather(father);
                                send.setHead(ip,recvLevel,father,recvLevel -1);
                                send.sendMessage(ADD_RS,str);
                        }
                        else if(recv_rs.getType() ==DELETE_RS){
                                nodes[recvLevel].deleteIndex(req_ip,filename);
                                nodes[recvLevel].getFather(father);
                                send.setHead(ip,recvLevel,father,recvLevel -1);
                                send.sendMessage(DELETE_RS,str);
                        }
			else{
				printf("node[%d] Receive a unknown message\n",recvLevel);
                                pthread_exit((void*)-1);;
			}
		}
		else if(recvLevel == getDepth() - 1){
		        while (nodes[recvLevel].getStatus() != STAT_NORMAL);
			if(recv_rs.getType() == SERVICE_REQ){
		//		load++;
        			link[filename]++;
        			printf("CLIENT FROM %s --> %s\n",req_ip,filename);
				
        			do_service(str);
				//printf("do_service() end\n");
        	//		load--;
        			link[filename]--;	
			}
			else if(recv_rs.getType() == ADD_RS){
				send_admin.setHead(ip,recvLevel,recv_rs.getSender(),-2);
				send_admin.sendMessage(SEND_PLEASE,"");
			    	receiveFile(filename);
#ifdef RUN
				printf("add file in leaf %s\n",ip.c_str());
#endif
				nodes[recvLevel].addIndex(req_ip,filename);
				nodes[recvLevel].mapToFile("source.txt");
				nodes[recvLevel].getFather(father);
				send.setHead(ip,recvLevel,father,recvLevel -1);
				send.sendMessage(ADD_RS,str);
			}
			else if(recv_rs.getType() ==DELETE_RS){
				nodes[recvLevel].deleteIndex(req_ip,filename);
				nodes[recvLevel].mapToFile("source.txt");
#ifdef RUN
				printf("delete file in leaf %s\n",ip.c_str());
#endif
				nodes[recvLevel].getFather(father);
                                send.setHead(ip,recvLevel,father,recvLevel -1);
                                send.sendMessage(DELETE_RS,str);
			}
			else if(recv_rs.getType() == LOAD_REQ){
				sprintf(str,"%d ",getLoad());
				send_admin.setHead(ip,recvLevel,recv_rs.getSender(),-2);
				send_admin.sendMessage(ANSWER_LOAD,str);
			}
			else{
				printf("node[%d] Receive a unknown message\n",recvLevel);
				pthread_exit((void*)-1);
			}
		}
		else{
		        while (nodes[recvLevel].getStatus() != STAT_NORMAL);
			if(recv_rs.getType() == FORWARD_REQ){
				if(nodes[recvLevel].has_source(filename)){
					//if(nodes[recvLevel].match_ip(req_ip)){
						nodes[recvLevel].sendSuccess(str);
						if(nodes[recvLevel].sendFirstChild(str) < 0){
							printf("node[%d] send first child error\n",recvLevel);
							pthread_exit((void*)-1);
						}
						printf("node[%d] send first child\n",recvLevel);
					//}
					/*else
						nodes[recvLevel].sendFail(str);*/
				}
				else 
					nodes[recvLevel].sendFail(str);
			}
			else if(recv_rs.getType() == FIND_SUCCESS){
			    printf("node[%d] receive FIND_SUCCESS\n",recvLevel);
				nodes[recvLevel].erase(str);	
			}
			else if(recv_rs.getType() == FIND_FAIL){
			    printf("node[%d] receive FIND_FAIL\n",recvLevel);
				if(nodes[recvLevel].sendNextChild(str) < 0){
					nodes[recvLevel].erase(str);
//					sleep(20);
					send_service.setHead(ip,recvLevel,req_ip,-1);
					send_service.sendMessage(WAIT_REQ,str);
				}
			}
			else if(recv_rs.getType() == ADD_RS){
                                nodes[recvLevel].addIndex(req_ip,filename);
                                nodes[recvLevel].getFather(father);
                                send.setHead(ip,recvLevel,father,recvLevel -1);
                                send.sendMessage(ADD_RS,str);
                        }
                        else if(recv_rs.getType() ==DELETE_RS){
                                nodes[recvLevel].deleteIndex(req_ip,filename);
                                nodes[recvLevel].getFather(father);
                                send.setHead(ip,recvLevel,father,recvLevel -1);
                                send.sendMessage(DELETE_RS,str);
                        }
			else{
				printf("node[%d] Receive a unknown message\n",recvLevel);
				pthread_exit((void*)-1);
			}
		}	
	}
}

int RealNode::sendFile(string ip,char *file){
        FILE* fp;
        int sockfd;
        char buf[4096];
        struct sockaddr_in servaddr;
        int n = 0;
	int tmp = 0;
	
        printf("begin to send filename: %s , %d to %s\n",file,strlen(file),ip.c_str());

        if((fp = fopen(file,"rb")) < 0){
                printf("open failed\n");
                return -1;
        }
        if ((sockfd = socket(AF_INET, SOCK_STREAM, 0)) < 0){
                printf("socket() error ");
                return -1;
        }
        bzero(&servaddr, sizeof(servaddr));
        servaddr.sin_family = AF_INET;
        servaddr.sin_port = htons(FTP_PORT);
        if (inet_pton(AF_INET, ip.c_str(), &servaddr.sin_addr) < 0){
                printf("inet_pton() error ");
                return -1;
        }
        if (connect(sockfd, (sockaddr *)&servaddr, sizeof(servaddr)) < 0){
                printf("connect() error\n");
                return -1;
        }

/*        if((send(sockfd,file,strlen(file),0))< 0 ){
                printf("send file name error\n");
                return -1;
        }
*/
        while((n = fread(buf,1,sizeof(buf),fp))>0){
//                printf("fread n = %d\n",n);
	    	tmp += n;
                if((n = send(sockfd, buf, n, 0)) < 0){
                        printf("send() error ");
                        return -1;
                }
//                else
//                        printf("send n = %d\n",n);
        }
	printf("%s( %d bytes ) send complete\n",file,tmp);

        close(sockfd);
        fclose(fp);
	//sleep(5);
        return 0;
}

int RealNode::receiveFile(char * filename){
        FILE* fp;
        char buf[4096];
        int listenfd,connfd,sockfd;
        struct sockaddr_in servaddr;
        //char filename[30];
        int n = 0;
	int tmp = 0;

        if((listenfd = socket(AF_INET,SOCK_STREAM,0)) < 0){
                printf("socket() error\n");
                return -1;
        }
        bzero(&servaddr,sizeof(servaddr));
        servaddr.sin_family = AF_INET;
        servaddr.sin_addr.s_addr = htonl(INADDR_ANY);
        servaddr.sin_port = htons(FTP_PORT);
        if((bind(listenfd,(sockaddr*)&servaddr,sizeof(servaddr))) < 0){
                printf("bind() error\n");
                return -1;
        }
        listen(listenfd,5);
        connfd = accept(listenfd,(sockaddr*)NULL,NULL);
        if((fp = fopen(filename,"wb")) < 0){
                printf("open file error\n");
                return -1;
        }

        while((n = recv(connfd,buf,4096,0)) > 0){
//                printf("receive n = %d\n",n);
	    tmp += n;
                 fwrite(buf,1,n,fp);
//                        printf("fwrite n = %d\n",n);
        }
	printf("%s ( %d bytes )received complete!\n",filename,tmp);
	
        close(connfd);
	//close(listenfd);
        fclose(fp);
        return 1;
}

/*void*
thread_service(void *arg)
{
    char ip[20], filename[50];

    //RealNode *temp = ((RealNode *)(((servArg *)arg)->REAL_ptr));
    
    ((RealNode *)arg)->addLoad();

    sscanf(((RealNode *)arg)->ip_file, "%s %s", ip, filename);
    //((RealNode *)(((servArg *)arg)->REAL_ptr))->sendFile(ip, filename);
    ((RealNode *)arg)->sendFile(ip, filename);
    //printf("sleep now\n");
    sleep(10);
    //printf("wake up\n");
    //((RealNode *)(((servArg *)arg)->REAL_ptr))->deleteLoad();
    ((RealNode *)arg)->deleteLoad();
}

void
RealNode::do_service(char *str)
{
    pthread_t tid;
    servArg servarg;
    void *tret;
    
    printf("leaf %s receive request\n", this->ip.c_str());
    //servarg.REAL_ptr = this;
    strcpy(this->ip_file, str);
    if (pthread_create(&tid, NULL, thread_service, (void *)this) != 0)
    {
	perror("pthread_create() error ");
	exit(0);
    }
*/    //printf("service end\n");
    /**if (pthread_join(tid, &tret) != 0)
    {
	perror("pthread_join() error ");
	exit(0);
    }*/
//}

void*
thread_service(void *arg)
{
    char ip[20],filename[50];
   
    pthread_detach(pthread_self());
    ((RealNode *)(((servArg *)arg)->REAL_ptr))->addLoad();
    sscanf(((servArg *)arg)->ip_file, "%s %s", ip, filename);
    ((RealNode *)(((servArg *)arg)->REAL_ptr))->sendFile(ip, filename);
    sleep(10);
    ((RealNode *)(((servArg *)arg)->REAL_ptr))->deleteLoad();
    free((servArg*)arg);
//    pthread_exit((void*)1);
}

void
RealNode::do_service(char *str)
{
    pthread_t tid;
    servArg* servarg;
    printf("leaf %s receive request\n",ip.c_str());
    servarg = (servArg*)malloc(sizeof(servArg));
    servarg->REAL_ptr = this;
    strcpy(servarg->ip_file,str);
    if(pthread_create(&tid,NULL,thread_service,(void*)servarg) != 0){
 	printf("%s\n", strerror(errno));
    	perror("pthread_create() error in do_service");
	return;
    }
}
